package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Stream_updateStateByKey {
  def main(args: Array[String]): Unit = {

    //如果SparkStreaming程序意外终止，那么重启后就需要从检查点恢复数据
    val ssc: StreamingContext = StreamingContext.getOrCreate("cp", () => {
      //TODO SparkStreaming环境
      val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
      //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
      val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
      ssc.checkpoint("cp")
      val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)
      //有状态数据操作
      //所谓的有状态，其实就是将采集周期的计算结果保存到检查点中。工作中都是保存在第三方的数据库中例如：Redis
      val wordDS: DStream[(String, Int)] = dstream.map((_, 1))
      //有装态数据操作的聚合需要自行完成
      wordDS.updateStateByKey(
        (seq: Seq[Int], opt: Option[Int]) => {
          Option(seq.sum + opt.getOrElse(0))
        }
      ).print()
      ssc
    })


    ssc.start()
    ssc.awaitTermination()
  }

}
